package com.example.framework.mq.rocketmq.core.service;

import com.example.framework.common.util.GenericUtils;
import com.example.framework.common.util.MsgPackUtils;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;

import java.util.List;

@Slf4j
public abstract class RocketMQConsumerService<T> implements MessageListenerConcurrently {

    @Override
    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
        String topic = msgs.getFirst().getTopic();
        try {
            for (MessageExt messageExt : msgs) {
                T msgBody = MsgPackUtils.deserialize(messageExt.getBody(), getType());
                log.info("线程{}收到新消息，topic={}, tag={}, msg={}", Thread.currentThread().getName(), messageExt.getTopic(), messageExt.getTags(), msgBody);
                consume(topic, msgBody);
            }
            // 标记该消息已经被成功消费
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        } catch (Exception e) {
            log.error("消费消息失败，Topic={}, msg={}, e={}", topic, msgs, e.getMessage());
            return ConsumeConcurrentlyStatus.RECONSUME_LATER;
        }
    }

    public abstract void consume(String topic, T msgBody);

    public Class<T> getType() {
        return GenericUtils.getGenericClass(getClass(), 0);
    }
}
